<?php
declare(strict_types=1);

namespace app\common\service;

use Exception;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use think\facade\Env;

/**
 * rabbitMQ类
 * @package app\common\service
 *
 * @Author: Li Guo Xin <guoxinlee129@gmail.com>
 * @Date: 2020/7/7
 * @Time: 9:26
 */
class RabbitMQ
{
    private static $exchange;#交换器名称
    private static $queue;#队列名称
    private static $exchangeType;#交换器类型
    private static $passive = false;#没有则创建交换器
    private static $autoDelete = false;#通道关闭后，队列自动删除
    private static $durable = true;#持久化
    private static $exclusive = true;#是否排外,限制一个队列只能有一个消费者
    private static $channel;
    private static $connection;
    private static $no_ack = false;

    public function setExchange($exchange): void
    {
        self::$exchange = $exchange;
    }

    public function setQueue($queue): void
    {
        self::$queue = $queue;
    }

    public function setPassive(): void
    {
        self::$passive = true;
    }

    public function __construct(
        $exchange = '',
        $queue = 'direct',
        $exchangeType = '',
        $passive = false,
        $autoDelete = false,
        $durable = true,
        $exclusive = true,
        $no_ack = true
    ) {
        self::$connection = new AMQPStreamConnection(
            Env::get('rabbitmq.host'),
            Env::get('rabbitmq.port'),
            Env::get('rabbitmq.user'),
            Env::get('rabbitmq.pass'),
            Env::get('rabbitmq.vhost')
        );

        self::$exchange     = $exchange;
        self::$queue        = $queue;
        self::$exchangeType = $exchangeType;
        self::$passive      = $passive;
        self::$autoDelete   = $autoDelete;
        self::$durable      = $durable;
        self::$exclusive    = $exclusive;
        self::$no_ack       = $no_ack;

        self::$channel = self::$connection->channel();
        self::$channel->queue_declare(
            self::$queue, self::$passive, self::$durable, self::$exclusive,
            self::$autoDelete);
    }

    public function publisher($argv): void
    {
        /*
            name: $exchange
            type: direct
            passive: false
            durable: true // the exchange will survive server restarts
            auto_delete: false //the exchange won't be deleted once the channel is closed.
        */
        self::$channel->exchange_declare(self::$exchange, self::$exchangeType, self::$passive, self::$durable,
            self::$autoDelete);

        self::$channel->queue_bind(self::$queue, self::$exchange);

        $messageBody = implode(' ', array_slice($argv, 1));
        $message     = new AMQPMessage($messageBody,
            array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
        self::$channel->basic_publish($message, self::$exchange);

        self::$channel->close();
        self::$connection->close();
    }

    /**
     * 消费者
     * @param $consumerTag
     * @param $no_local
     * @param  bool  $no_wait
     * @throws \ErrorException
     * @Author: Li Guo Xin <guoxinlee129@gmail.com>
     * @Date: 2020/7/9
     * @Time: 14:52
     */
    public function consumer($consumerTag,$no_local,$no_wait = false): void
    {
        self::$channel->queue_bind(self::$queue, self::$exchange);

        /**
         * consumerTag 消费者标识符
         * no_local 不接收此使用者发布的消息
         * no_ack 不需要手动确认
         * nowait 不等待
         * callback 回调方法
         */
        self::$channel->basic_consume(self::$queue, $consumerTag, $no_local, self::$no_ack, self::$exclusive, $no_wait,
            'process_message');

        register_shutdown_function('shutdown', self::$channel, self::$connection);

        //只要通道已注册回调,就循环
        while (self::$channel->is_consuming()) {
            self::$channel->wait();
        }
    }

    /**
     * @param  AMQPMessage  $message
     */
    public function process_message($message): void
    {
        echo "\n--------\n";
        echo $message->body;
        echo "\n--------\n";

        $message->ack();

        // Send a message with the string "quit" to cancel the consumer.
        if ($message->body === 'quit') {
            $message->getChannel()->basic_cancel($message->getConsumerTag());
        }
    }

    /**
     * PHP终止时执行
     * @param $channel
     * @param $connection
     *
     * @Author: Li Guo Xin <guoxinlee129@gmail.com>
     * @Date: 2020/7/9
     * @Time: 14:55
     */
    public function shutdown($channel,$connection): void
    {
        $channel->close();
        $connection->close();
    }
}